package io.kiki.sba.registry.remoting.http;

import io.kiki.sba.registry.api.Callback;
import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.remoting.http.jetty.server.HttpConnectionCustomFactory;
import lombok.Getter;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.CustomRequestLog;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.Slf4jRequestLogWriter;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Slf4jLog;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.glassfish.jersey.internal.guava.ThreadFactoryBuilder;
import org.glassfish.jersey.jetty.JettyHttpContainer;
import org.glassfish.jersey.jetty.internal.LocalizationMessages;
import org.glassfish.jersey.process.JerseyProcessingUncaughtExceptionHandler;
import org.glassfish.jersey.server.ContainerFactory;
import org.glassfish.jersey.server.ResourceConfig;
import org.glassfish.jersey.server.spi.Container;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.ws.rs.ProcessingException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;


public class JerseyJettyServer implements Server {
    private static final String NCSA_FORMAT = "%{client}a \"%r\" %s %O";
    private static final Logger logger = LoggerFactory.getLogger(JerseyJettyServer.class);

    private final ResourceConfig resourceConfig;


    @Getter
    private final URI baseUri;

    private final AtomicBoolean isStarted = new AtomicBoolean(false);
    private org.eclipse.jetty.server.Server server;


    public JerseyJettyServer(ResourceConfig resourceConfig, URI baseUri) {
        this.resourceConfig = resourceConfig;
        this.baseUri = baseUri;
    }

    public static org.eclipse.jetty.server.Server createServer(final URI uri, final ResourceConfig resourceConfig, final boolean start) {
        if (uri == null) {
            throw new IllegalArgumentException(LocalizationMessages.URI_CANNOT_BE_NULL());
        }

        JettyHttpContainer handler = ContainerFactory.createContainer(JettyHttpContainer.class, resourceConfig);

        int defaultPort = Container.DEFAULT_HTTP_PORT;

        final int port = (uri.getPort() == -1) ? defaultPort : uri.getPort();

        final org.eclipse.jetty.server.Server server = new org.eclipse.jetty.server.Server(new JettyConnectorThreadPool());
        // init requestLog
        Slf4jRequestLogWriter writer = new Slf4jRequestLogWriter();
        CustomRequestLog log = new CustomRequestLog(writer, NCSA_FORMAT);
        server.setRequestLog(log);
        final ServerConnector http = new ServerConnector(server, new HttpConnectionCustomFactory());
        http.setAcceptQueueSize(512);
        http.setPort(port);
        server.setConnectors(new Connector[]{http});
        if (handler != null) {
            server.setHandler(handler);
        }

        if (start) {
            try {
                // Start the server.
                server.start();
            } catch (Throwable e) {
                throw new ProcessingException(LocalizationMessages.ERROR_WHEN_CREATING_SERVER(), e);
            }
        }
        return server;
    }


    public void startServer() {
        if (isStarted.compareAndSet(false, true)) {
            try {
                Log.setLog(new Slf4jLog());
                server = createServer(getBaseUri(), resourceConfig, true);
            } catch (Throwable e) {
                isStarted.set(false);
                logger.error("Start Jetty jersey server error!", e);
                throw new RuntimeException("Start Jetty jersey server error!", e);
            }
        }
    }

    @Override
    public boolean isOpen() {
        if (server != null) {
            return server.isStarted();
        }
        return false;
    }

    @Override
    public List<Channel> getChannelList() {
        return Collections.emptyList();
    }

    @Override
    public Map<String, Channel> selectAvailableChannelsForHostAddress() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Map<String, List<Channel>> selectAllAvailableChannelsForHostAddress() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Channel getChannel(InetSocketAddress remoteAddress) {
        return null;
    }

    @Override
    public Channel getChannel(Url url) {
        return null;
    }

    @Override
    public void close(Channel channel) {
        throw new UnsupportedOperationException("Jersey Server don't support close Channel.");
    }

    @Override
    public InetSocketAddress getLocalAddress() {
        return new InetSocketAddress(getBaseUri().getPort());
    }

    @Override
    public void close() {
        if (server != null) {
            try {
                server.stop();
            } catch (Throwable e) {
                logger.error("Jersey Jetty Server stop error!", e);
                throw new RuntimeException("Jersey Jetty Server stop error!", e);
            }
            return;
        }
        throw new RuntimeException("Jersey Server has not started!Server Channel has not created!");
    }

    @Override
    public boolean isClosed() {
        if (server != null) {
            return server.isStopped();
        }
        return true;
    }

    @Override
    public Object sendSync(Channel channel, Object message, int timeoutMillis) {
        return null;
    }

    @Override
    public void sendCallback(Channel channel, Object message, Callback callback, int timeoutMillis) {
    }

    @Override
    public int getChannelCount() {
        return 0;
    }

    private static final class JettyConnectorThreadPool extends QueuedThreadPool {
        private final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("jetty-http-server-%d").setUncaughtExceptionHandler(new JerseyProcessingUncaughtExceptionHandler()).build();

        @Override
        protected Thread newThread(Runnable runnable) {
            return threadFactory.newThread(runnable);
        }
    }
}
